package com.hhe.datax_transfer.service;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.NioUtil;
import cn.hutool.core.util.StrUtil;
import com.hhe.datax_transfer.cache.DataxJobMonitoringControl;
import com.hhe.datax_transfer.cache.ExecuteJobsQueue;
import com.hhe.datax_transfer.model.DataxMonitoringControlModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;

/**
 * 异步执行类
 *
 * @author HHe
 * @version V1.0
 * @date 2021/1/13 12:11
 */
@Slf4j
@Component
public class AsyncExecuteTransferOperation {
    @Value("${datax.path}")
    private String dataxPath;

    @Async("taskExecutor")
    public void test() {
        log.info("发送消息");
        System.out.println("子线程名称：" + Thread.currentThread().getName());
    }

    /**
     * 执行DataX配置
     *
     * @param fullPathName 配置文件全路径名
     * @version V1.0
     * @author HHe
     * @date 2021/1/13 14:30
     */
    @Async("taskExecutor")
    public void executeDataXJson(String fullPathName) {
        //队列任务开始执行
        DataxJobMonitoringControl.execute(fullPathName);
        long staTime = System.currentTimeMillis();
        long runTime;
        String status;
        try {
            //当前线程
            Thread thread = Thread.currentThread();
            DataxJobMonitoringControl.updateJobMess(new DataxMonitoringControlModel(fullPathName, thread));
            Process process = Runtime.getRuntime().exec("python " + dataxPath + "/bin/datax.py " + fullPathName);
//            long c1Time = runTime - staTime;
//            String runingLog = fullPathName+" | "+staTime+"->"+runTime+" | "+c1Time;
//            DataxJobMonitoringControl.putRuningLog(runingLog);
//            log.info("运行命令耗时：{}", c1Time);
            //写入文件
            InputStream inputStream = process.getInputStream();
            //替换文件夹的路径h和后缀
            String logFilePath = fullPathName.replace("/datax_json_lib/", "/datax_json_logs/").replace(".json", ".log");
            String logFolderPath = StrUtil.subBefore(logFilePath, "/", true);
            File logFolder = new File(logFolderPath);
            if (!logFolder.exists()) {
                logFolder.mkdirs();
            }
            File logFile = new File(logFilePath);
            if (!logFile.exists()) {
                logFile.createNewFile();
            }
            FileOutputStream fileOutputStream = new FileOutputStream(logFile);
            NioUtil.copyByNIO(inputStream, fileOutputStream, 1024 << 4, null);
            fileOutputStream.close();
            inputStream.close();
            //完成任务
            runTime = System.currentTimeMillis();
            status = "success";
            DataxJobMonitoringControl.executeFinish(fullPathName);
        } catch (IOException e) {
            runTime = System.currentTimeMillis();
            status = "failure";
            DataxJobMonitoringControl.executeFinish(fullPathName);
            log.error(e.getMessage());
        }
        long c1Time = runTime - staTime;
        String runingLog = status + " | [" + DateUtil.formatDateTime(DateUtil.date(staTime)) + "] -> [" + DateUtil.formatDateTime(DateUtil.date(runTime)) + "] | " + c1Time + "ms";
        DataxJobMonitoringControl.putRuningLog(fullPathName);
        DataxJobMonitoringControl.putRuningLog(runingLog);
        log.info("运行命令耗时：{}", c1Time);
    }
}
